package com.millstein.realtime.app.dwd.log;

import com.alibaba.fastjson.JSONObject;
import com.millstein.realtime.app.base.BaseAppV1;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.CommonUtil;
import com.millstein.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;


/**
 * @Description
 * @Author tsing
 * @Date 2024-09-30 11:11
 */
public class Dwd_02_TrafficUniqueVisitorDetail extends BaseAppV1 {

    public static void main(String[] args) {
        new Dwd_02_TrafficUniqueVisitorDetail().init(
                3002,
                3,
                "Dwd_02_TrafficUniqueVisitorDetail",
                Constants.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env          执行环境对象
     * @param streamSource 待处理的数据流
     */
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> streamSource) {
        // 1.过滤出每天每个用户的第一条访问日志
        SingleOutputStreamOperator<String> dataStream = filterFirstDataByUidAndDay(streamSource);

        // 2.将数据写入kafka
        dataStream.addSink(FlinkSinkUtil.getKafkaSink(Constants.TOPIC_DWD_TRAFFIC_UV_DETAIL));
    }

    private SingleOutputStreamOperator<String> filterFirstDataByUidAndDay(
            DataStreamSource<String> streamSource
    ) {
        return streamSource
                .map(JSONObject::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("ts");
                                    }
                                }))
                .keyBy(json -> json.getJSONObject("common").getString("uid"))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<JSONObject, String, String, TimeWindow>() {
                    // 记录每个用户最后一次访问的日期
                    private ValueState<String> visitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        visitDateState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("visit_date_state", String.class)
                        );
                    }

                    @Override
                    public void process(
                            String uid,
                            Context context,
                            Iterable<JSONObject> elements,
                            Collector<String> out
                    ) throws Exception {
                        // 获取该用户最后访问的日期
                        String visitDate = visitDateState.value();
                        // 获取当前窗口的日期
                        String today = CommonUtil.formatDate(context.window().getStart());
                        // 如果两个日期不一致，那么可以认为日期已经切换，该窗口就是日期切换后的第一个窗口
                        // 再从这个窗口中取出ts最小的数据作为当天该用户的独立访客数据
                        // 最后更新最后访问日期为当天
                        if (!today.equals(visitDate)) {
                            List<JSONObject> list = CommonUtil.toList(elements);
                            JSONObject min = Collections.min(list, Comparator.comparing(o -> o.getLong("ts")));
                            out.collect(min.toJSONString());
                            visitDateState.update(today);
                        }
                    }
                });
    }
}
